AWS Lambda Powertoolsが便利すぎた #serverless #python
こんにちは、クラスメソッドの岡です。
この記事は AWS LambdaとServerless Advent Calendar 2020 の7日目の記事です。
AWS Lambda Powertoolsとは?
Lambdaでの実装をサポートしてくれるライブラリです。 現在、ライブラリが提供されているのはPythonとJavaの2つになります。
ちなみに、DAZNからNode.js用の DAZN Lambda Powertools もでています。
動作確認環境
- Python: 3.8.5
- aws-lambda-powertools-python: 1.8.0
- Serverless Framework: 2.15.0
主な機能
- Logging
- LambdaのContextを埋め込んだログの構造化
- Tracing
- X-Rayでのトレース
- Metrics
- CloudWatchのカスタムメトリクスの作成
- Utilities
- バリデーションやパラメータ取得などLambdaでよく使う処理をラップしたもの
インストール
$ pip install aws-lambda-powertools
Logging
ログの設定のため、以下2つの環境変数を設定しておきます。
- POWERTOOLS_SERVICE_NAME: sample-devio-app
- LOG_LEVEL: DEBUG
Lambdaのハンドラに設定する関数に対して、inject_lambda_context
のデコレーターを付与するだけで、ログのJSONフォーマット化とContextの埋め込みをやってくれます。
import json from typing import Any, Dict from aws_lambda_powertools import Logger from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent logger = Logger() @logger.inject_lambda_context(log_event=True) def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]: logger.debug('debug') logger.info('info') return { 'statusCode': 200, 'body': json.dumps({'message': 'Success.'}) }
log_event
パラメータをTrueにすることで、eventデータをINFOログで自動的に出力してくれます。
{ "level": "INFO", "location": "decorate:245", "message": { "limit": 3 }, "timestamp": "2020-12-07 08:00:00,000", "service": "sample-devio-app", "sampling_rate": 0, "cold_start": false, "function_name": "list_items", "function_memory_size": "1024", "function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items", "function_request_id": "20014cb4-70cb-470d-beb2-839ce13b618e", "xray_trace_id": "1-5fcbb424-02fdaa386ca103a26e858647" }
実際に出力されたeventデータです。構造化されているのでCloudWatch Logs Insightsで簡単にログを探せるようになりますね。
serviceの部分には環境変数 POWERTOOLS_SERVICE_NAME
が設定されています。
また、型ヒントを取り入れている場合は utilities
のLambdaContextとAPIGatewayProxyEvent(Lambda
プロキシ統合イベント)が提供されているので、eventとcontextのアノテーションに使うことができます。
Handlerモジュール以外でログ出力する
Loggerのインスタンス生成時にChildパラメータをTrueに設定します。
from aws_lambda_powertools import Logger logger = Logger(child=True) class Auth: def verify(self, token: string): logger.info(token) # 中略 return result
上記のようにすることでLoggerを初期化せずにコード全体で共通して利用できます。
パラメータを追加する
Loggerをstructure_logs
関数を使うことで、構造化したログに簡単に任意のパラメータを追加できます。
試しにAPI Gatewayの認証から渡される principalId
をuser_idとしてログにセットします。
def set_logs(self) -> None: logger.structure_logs(append=True, user_id=self.principal_id) # principal_id = user_id logger.info('test') def __set_principal_id(self, event: APIGatewayProxyEvent]) -> None: self.principal_id = event.get( 'requestContext', {}).get( 'authorizer', {}).get('principalId')
{ "level": "INFO", "location": "set_logs:59", "message": "test", "timestamp": "2020-12-07 08:00:00,000", "service": "sample-devio-app", "sampling_rate": 0, "cold_start": false, "function_name": "list_items", "function_memory_size": "1024", "function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items", "function_request_id": "20014cb4-70cb-470d-beb2-839ce13b618e", "user_id": "test_user", "xray_trace_id": "1-5fcbb424-02fdaa386ca103a26e858647" }
"user_id": "test_user"
が追加されました。
locationにはログ出力している関数名が出力されます。
pytest
inject_lambda_context
デコレータを追加したハンドラーをそのままテストすると構造化の処理でコケてしまうので、pytest実行時には前処理でダミーのcontextを生成して渡してあげる必要があります。
@pytest.fixture def lambda_context(): lambda_context = { "function_name": "list_items", "memory_limit_in_mb": 128, "invoked_function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:list_items", "aws_request_id": "20b4014c-beb2-839ce70cb-470d-13b618e", } return namedtuple("LambdaContext", lambda_context.keys())(*lambda_context.values()) def test_list_items(handler, lambda_context): test_event = {'test': 'event'} handler(test_event, lambda_context) # this will now have a Context object populated
Trace
Tracerを使うことで、aws_xray_sdkを個別に使わずにサブセグメントやアノテーションの定義を簡単に記述できます。(X-Rayの基本的な説明はここでは割愛します)
今回はServerless Frameworkでデプロイしているのでserverless.ymlにX-Rayのトレースの設定を追加していきます。
provider: name: aws region: ap-northeast-1 tracing: apiGateway: true lambda: true iamRoleStatements: - Effect: Allow Action: - logs:CreateLogGroup - logs:CreateLogStream - logs:PutLogEvents - dynamodb:DescribeTable - dynamodb:Query - dynamodb:Scan - dynamodb:GetItem - dynamodb:PutItem - dynamodb:UpdateItem - dynamodb:DeleteItem - xray:PutTraceSegments - xray:PutTelemetryRecords Resource: "*"
handlerに capture_lambda_handler
デコレータを追加します。
from aws_lambda_powertools import Logger from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent from aws_lambda_powertools import Tracer tracer = Tracer() logger = Logger() @tracer.capture_lambda_handler @logger.inject_lambda_context(log_event=True) def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]: logger.debug('debug') logger.info('info') return { 'statusCode': 200, 'body': {'message': 'Success.'} }
DynamoDBからデータ取得するLambdaを直実行した場合のTraceです。
アノテーションの追加
トレースをフィルタするアノテーションを追加します。
from aws_lambda_powertools import Tracer tracer = Tracer() @tracer.capture_lambda_handler def handler(event, context): tracer.put_annotation(key="PaymentStatus", value="SUCCESS")
TracerはHanderモジュール以外でインスタンス化した場合も既存構成を引き継いでくれます。
メタデータの追加
from aws_lambda_powertools import Tracer tracer = Tracer() @tracer.capture_lambda_handler def handler(event, context): ret = some_logic() tracer.put_metadata(key="payment_response", value=ret)
並列リクエストのトレース
aiohttp を使った並列リクエストのトレースも可能です。
import asyncio import aiohttp from aws_lambda_powertools import Tracer from aws_lambda_powertools.tracing import aiohttp_trace_config tracer = Tracer() async def aiohttp_task(): async with aiohttp.ClientSession(trace_configs=[aiohttp_trace_config()]) as session: async with session.get("https://httpbin.org/json") as resp: resp = await resp.json() return resp
pytest
ユニットテストの際には環境変数で無効化できます。
$ POWERTOOLS_TRACE_DISABLED=1 python -m pytest
Utilities
Event Source Data Classes
今回はAPI GatewayのLambdaプロキシ統合を利用する想定なので APIGatewayProxyEvent
を指定していますが、他にも以下のイベントソースを指定できます。
import json from typing import Any, Dict from aws_lambda_powertools import Logger from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent logger = Logger() @logger.inject_lambda_context(log_event=True) def handler(event: APIGatewayProxyEvent, context: LambdaContext) -> Dict[str, Any]: logger.debug('debug') logger.info('info') return { 'statusCode': 200, 'body': json.dumps({'message': 'Success.'}) }
- API Gateway Proxy
- API Gateway Proxy event v2
- CloudWatch Logs
- Cognito User Pool
- DynamoDB streams
- EventBridge
- Kinesis Data Stream
- S3
- SES
- SNS
- SQS
Parameters
Parametersユーティリティを使って、以下からパラメータを取得してキャッシュまでしてくれます。
- SSM Parameter Store
- Secrets Manager
- DynamoDB
SSM Parameter Storeから取得
from aws_lambda_powertools.utilities import parameters def handler(event, context): value = parameters.get_parameter("/my/parameter") values = parameters.get_parameters("/my/path/prefix") for k, v in values.items(): logger.debug(f"{k}: {v}")
(boto3.ssm の get_parameter, get_parameters_by_pathをラップ)
Secrets Managerから取得
from aws_lambda_powertools.utilities import parameters def handler(event, context): value = parameters.get_secret("my-secret")
(boto3.secretsmanager の get_secret_valueをラップ)
DynamoDBから取得
from aws_lambda_powertools.utilities import parameters dynamodb_provider = parameters.DynamoDBProvider(table_name="my-table") def handler(event, context): value = dynamodb_provider.get("my-parameter")
(boto3.resource.dynamodb の get_itemをラップ)
from aws_lambda_powertools.utilities import parameters dynamodb_provider = parameters.DynamoDBProvider( table_name="my-table", key_attr="MyKeyAttr", sort_attr="MySortAttr", value_attr="MyvalueAttr" ) def handler(event, context): value = dynamodb_provider.get("my-parameter")
(boto3.resource.dynamodb の queryをラップ)